#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#include <pthread.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <sys/epoll.h>
#include <assert.h>
#include <arpa/inet.h>
#include <stdarg.h>
#include <signal.h>

#include <kvsp.h>
#include <kvs.h>

#include "server_func.h"

#define MAXEVENTS			1024
#define MAXEPOLLS			128
#define	SYNC_TIME_SEC			600
#define	SERVER_LAZY_TIME_HORIZON_MSEC	(300*1000)

pthread_mutex_t critical_source_mutex=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t	thread_func_cond=PTHREAD_COND_INITIALIZER;
pthread_mutex_t thread_func_mutex=PTHREAD_MUTEX_INITIALIZER;
char* kvs_image=NULL;


static void thread_func();
static void sig_usr(int);

int main(int argc, char** argv)
{
	int listen_fd, connect_fd, epoll_fd, nfds;
	int ret;
	struct epoll_event ev, events[MAXEVENTS];
	EHPTR ehptr;

	pthread_t t_id;
	pthread_mutex_init(&critical_source_mutex, NULL);
	pthread_mutex_init(&thread_func_mutex, NULL);
	pthread_cond_init(&thread_func_cond, NULL);

	if (signal(SIGUSR1, sig_usr)==SIG_ERR) {
		printf ("can't catch SIGUSR1");
	}

/*********************************/
	if (argc!=4)
	{
		printf ("Usage:\n");
		printf ("~/kvs$ ./server PORT BigfilePathname INIT_MOD\n");
		printf ("PORT: specify listen port.\n");
		printf ("BigfilePathname: specify the bigfile location.\n");
		printf ("INIT_MOD: 'L'or'l' for load, 'C'or'c' for create.\n");
		printf ("\n");
		printf ("an example: ./server 8080 /tmp/kvsdisk c\n");
		printf ("Create a new kvs instance (a big file) at /tmp/kvsdisk, listen port is 8080.\n\n");
		return 0;
	}
	
	
	int listen_port;
	char pathname[1024];
	char init_mod[1024];
	int init_type;
	memset(pathname, 0, sizeof(pathname));
	memset(init_mod, 0, sizeof(init_mod));

	sscanf(argv[1], "%d", &listen_port);
	sscanf(argv[2], "%s", pathname);
	sscanf(argv[3], "%s", init_mod);
	printf ("listen_port:%d\tpathname:%s\tinit_mod:%s\n",listen_port, pathname, init_mod);
	

	if (init_mod[0]=='c' || init_mod[0]=='C') {
		init_type=KVS_CREATE;
	} else {
		init_type=KVS_LOAD;
	}

	init(pathname,init_type, &kvs_image);
	
	if((listen_fd = deal_listen(listen_port))==-1)
		exit(-1);

	if ((epoll_fd = epoll_create(MAXEPOLLS)) < 0) {
		perror("epoll_create error");
		exit(-1);
	}

	ev.events = EPOLLIN;
	ev.data.fd = listen_fd;

	if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ev) == -1) {
		perror("epoll_ctl add error");
		exit(-1);
	}

/* start a new thread. */
	ret=pthread_create(&t_id, NULL, (void*)thread_func, NULL);

/*XXX: to do: ways of call "pthread_cond_signal()"
 *1.TimeOut (not only)
 *2.PUT many times.
 *3.It is a long time not run "thread_cond_signal",  but SYNC_TIME_SEC considered.
 */
	while (1)
	{
		int i;
		if ((nfds = epoll_wait(epoll_fd, events, MAXEVENTS, SERVER_LAZY_TIME_HORIZON_MSEC)) == -1) {
			perror("epoll_wait error");
			exit(-1);
		}

		if (nfds==0) {
			/* timeout */
			printf ("time out. server is lazy.\n");
			pthread_cond_signal(&thread_func_cond); /* if thread_func is not read, this will NOT blocked. */
			continue;
		}

		for (i = 0; i < nfds; i++)
		{
			if (events[i].data.fd == listen_fd) {
				slog(stderr, "EVENT FROM listenfd\n");
				connect_fd = deal_accept(listen_fd);
				ehptr = (EHPTR)malloc(sizeof(EH));
				assert(ehptr);
				init_event_handler(ehptr, connect_fd, epoll_fd);
				epoll_add(ehptr, EPOLLIN);
			} else if ((events[i].events&EPOLLHUP)||(events[i].events&EPOLLERR)) {
				pthread_mutex_lock(&critical_source_mutex);
				ehptr=(EHPTR)events[i].data.ptr;
				slog(stderr, "clientfd-%d closed!\n", ehptr->connect_fd);
				epoll_del(ehptr);
				pthread_mutex_unlock(&critical_source_mutex);
			} else if (events[i].events&EPOLLIN) {
				pthread_mutex_lock(&critical_source_mutex);
				ehptr = (EHPTR)events[i].data.ptr;
				if (ehptr->state==COMMAND_RECV) {
					slog(stderr, "Read command from cfd-%d\n",ehptr->connect_fd);
					com_parse(ehptr);
				} else if (ehptr->state==DATA_RECV) {
					slog(stderr, "Read data from cfd-%d %d of %d\n", ehptr->connect_fd,ehptr->buffpos,ehptr->buffsize);
					data_recv(ehptr);
				}
				pthread_mutex_unlock(&critical_source_mutex);
			} else if (events[i].events&EPOLLOUT) {
				pthread_mutex_lock(&critical_source_mutex);
				ehptr=(EHPTR)events[i].data.ptr;
				if (ehptr->state==REPLY_SEND) {
					slog(stderr, "Reply to cfd-%d\n", ehptr->connect_fd);
					reply_send(ehptr);
				} else if (ehptr->state==DATA_SEND) {
					slog(stderr, "Send data to cfd-%d %d of %d\n", ehptr->connect_fd,ehptr->buffpos,ehptr->buffsize);
					data_send(ehptr);
				}
				pthread_mutex_unlock(&critical_source_mutex);
			}
		}
	}  /* end of while(1) */
	return 0;
}

static void sig_usr(int signo)
{
	if (signo==SIGUSR1) {
		printf ("exit..\n");
		kv_exit();
	}
	return;
}


static void thread_func()
{
	int ret;
	struct timespec sync_time;
	char *p=NULL;
	
	while (1)
	{
		pthread_mutex_lock(&thread_func_mutex);
		sync_time.tv_sec=time(NULL)+SYNC_TIME_SEC;
		sync_time.tv_nsec=0;
		ret=pthread_cond_timedwait(&thread_func_cond, &thread_func_mutex, (const struct timespec*)&sync_time);
		if (ret!=0) {
			printf ("time out\n");
		} else {
			printf ("a condition send, system is lazy.\n");
		}

		/*FIXME:
		 * Another method without p!=NULL
		 * Use [last_time-cur_time>time] to check whether "kv_image_flush()" too frequently
		 */
		if (p!=NULL) {
			printf ("Still SYNCing...\n");
			pthread_mutex_unlock(&thread_func_mutex);
			continue;
		}

		{
			pthread_mutex_lock(&critical_source_mutex);
			p=malloc(KVS_IMAGE_SIZE);
			if (p==NULL) {
				printf ("malloc failed:%s,%d.\n",__FILE__, __LINE__);
			}
			memcpy(p, kvs_image, KVS_IMAGE_SIZE);
			pthread_mutex_unlock(&critical_source_mutex);

			kv_image_flush(p);
			
/* This is just test a LONG-TIME "kv_image_flush()"
			while (1)
			{
				;
			}
*/
			
			free(p);
			p=NULL;
		}

		pthread_mutex_unlock(&thread_func_mutex);
	}
}

